package cluster

import (
	"context"
	"encoding/json"
	"fmt"
	"lnzw/engine/etcd"
	"lnzw/engine/log"
	"reflect"
	"strconv"
	"strings"

	"github.com/coreos/etcd/mvcc/mvccpb"
	c "go.etcd.io/etcd/clientv3"
	"go.uber.org/zap"
)

type putWatch func(node INode)
type delWatch func(nodeType int32, nodeId int32)

type etcdDiscovery struct {
	pwatch putWatch
	dwatch delWatch
}

func (d *etcdDiscovery) Init() {
	etcd.Setup()
}

func (d *etcdDiscovery) Register(node INode) error {
	key := fmt.Sprintf(CLUSTER_FULL_PREFIX, node.NodeType(), node.NodeId())
	leaseRsp, err := etcd.GlobalEtcd.Lease.Grant(context.Background(), 5)
	if err != nil {
		log.Logger.Error("etcd grant failed", zap.Error(err), zap.String("key", key))
		return err
	}
	leaseId := leaseRsp.ID
	value, err := node.Marshal()
	if err != nil {
		return err
	}
	etcd.GlobalEtcd.Put(context.Background(), key, string(value), c.WithLease(leaseId))
	//续约操作
	go func() {
		c, err := etcd.GlobalEtcd.KeepAlive(context.Background(), leaseId)
		if err != nil {
			log.Logger.Error("etcd keepalive failed", zap.Error(err))
			return
		}
		for {
			<-c
		}
	}()
	log.Info("etcd register success", zap.String("key", key))
	return nil
}

func (d *etcdDiscovery) Discovery(svrs ...int32) ([]INode, error) {
	nodes := make([]INode, 0)
	for _, svr := range svrs {
		key := fmt.Sprintf("/server/%d", svr)
		getRsp, err := etcd.GlobalEtcd.Get(context.Background(), key, c.WithPrefix())
		//_, err := etcd.GlobalEtcd.Get(context.Background(), key, c.WithPrefix())
		if err != nil {
			log.Logger.Error("etcd get failed", zap.Error(err), zap.String("key", key))
			return nil, err
		}

		//etcd查询服务器
		for _, kv := range getRsp.Kvs {
			nodeByte := kv.Value
			nodeOrdinary := nodeType2Node[svr]
			node := reflect.New(reflect.ValueOf(nodeOrdinary).Elem().Type()).Interface().(INode)
			err = json.Unmarshal(nodeByte, node)
			if err != nil {
				log.Logger.Error("unmarshal failed", zap.Error(err), zap.String("key", key))
				return nil, err
			}
			err = node.UnMarshal()
			if err != nil {
				log.Logger.Error("unmarshal detail failed", zap.Error(err), zap.String("key", key))
				continue
			}
			nodes = append(nodes, node)
			log.Logger.Info("etcd get", zap.String("key", string(kv.Key)), zap.String("value", string(kv.Value)))
		}
		c := etcd.GlobalEtcd.Watch(context.Background(), key, c.WithPrefix())
		//监听服务
		go func() {
			for resp := range c {
				for _, ev := range resp.Events {
					sArr := strings.Split(string(ev.Kv.Key), "/")
					svr := sArr[2]
					typeStr, err := strconv.Atoi(svr)
					if err != nil {
						log.Logger.Error("parse failed", zap.Error(err), zap.String("svr", svr))
						continue
					}
					nodeType := int32(typeStr)
					if ev.Type == mvccpb.PUT {
						nodeByte := ev.Kv.Value

						nodeOrdinary := nodeType2Node[nodeType]
						node := reflect.New(reflect.ValueOf(nodeOrdinary).Elem().Type()).Interface().(INode)
						err = json.Unmarshal(nodeByte, node)
						if err != nil {
							log.Logger.Error("unmarshal failed", zap.Error(err), zap.String("key", key))
							continue
						}
						err = node.UnMarshal()
						if err != nil {
							log.Logger.Error("unmarshal detail failed", zap.Error(err), zap.String("key", key))
							continue
						}
						d.pwatch(node)
					} else {
						nodeIdStr := sArr[3]
						nodeId, err := strconv.Atoi(nodeIdStr)
						if err != nil {
							log.Logger.Error("parse failed", zap.Error(err), zap.String("nodeId", nodeIdStr))
							continue
						}
						d.dwatch(nodeType, int32(nodeId))
					}
					log.Logger.Info("etcd watch", zap.String("type", string(ev.Type)), zap.String("key", string(ev.Kv.Key)), zap.String("value", string(ev.Kv.Value)))
				}
			}
		}()
	}
	log.Info("etcd discovery success", zap.Any("nodes", nodes))
	return nodes, nil
}

func (d *etcdDiscovery) UnRegister(node INode) error {
	key := fmt.Sprintf(CLUSTER_FULL_PREFIX, node.NodeType(), node.NodeId())
	delRsp, err := etcd.GlobalEtcd.Delete(context.Background(), key)
	if err != nil {
		log.Logger.Error("etcd delete failed, key not exsit", zap.String("key", key))
		return err
	}
	if delRsp.Deleted == 0 {
		log.Logger.Warn("etcd delete failed, key not exsit", zap.String("key", key))
	}
	return nil
}
